/**
 * Copyright (C) 2009+ Spark Zheng
 *
 * @file        src/reactor.c 
 * @date        2010-04-05
 * @brief	server_framework
 * @see		dserver       
 * @history
 */

#include <unistd.h>
#include <dlib/str.h>
#include <dlib/mempool.h>

#include "utils.h"
#include "reactor.h"
#include "protocol.h"
#include "connection.h"
#include "prot_types.h"

static void _reactor_pipe_handler(int fd, short flag, void *arg);
static inline int _reactor_run_protocol(prot_node_t *prot);

static void _reactor_pipe_handler(int fd, short flag, void *arg)
{
	int size;
	long conn_addr;
	dserver_conn_t *conn = NULL;
	prot_node_t *prot = NULL;
	char pipe_buf[REACTOR_PIPE_BUF_SIZE] = {0};

	UNUSED_PARAM(flag);
	UNUSED_PARAM(arg);

	DLOG_DEBUG(LOG_MAIN, "EVENT(pipe PROT_IGNORE): recieved fd[%d].", fd);

	if ((size = read(fd, pipe_buf, REACTOR_PIPE_BUF_SIZE)) < 0) {
		DLOG_ERR(LOG_MAIN, "pipe-fd[%d] read error: %s", fd, strerror(errno));
		return;
	}

	/* get the conn struct address and protocol_node */
	sscanf(pipe_buf, "%ld", &conn_addr);
	conn = (dserver_conn_t *)conn_addr;
	prot = conn->c_prot_node;

	/* do connection cleanup */
	if (prot->p_ops.p_server_conn_exit != NULL) {
		(void)(*prot->p_ops.p_server_conn_exit)(conn);
	}

	return; 
}

static inline int _reactor_run_protocol(prot_node_t *prot)
{
	int ret = RET_SUCCESS;

	if (prot->p_ops.p_server_init != NULL) {
		/** server init, create sfd */
		prot->p_sfd = (*prot->p_ops.p_server_init)(prot->p_settings.p_host,
				prot->p_settings.p_port, prot->p_settings.p_timeout);
		if (prot->p_sfd < 0) {
			DLOG_ERR(LOG_MAIN, "protocol[%s] p_server_init error.", prot->p_name);
			return -E_SOCKET;
		}
	} else {
		DLOG_ERR(LOG_MAIN, "protocol[%s] p_server_init not set.", prot->p_name);
		return -E_INPUT_ILLEGAL;
	}
	
	/* TODO: rcv/snd buf, tcp/udp set socket options */
	//(void)dserver_set_rcv_timeout(prot->p_sfd, prot->p_settings.p_timeout, 0);
	//(void)dserver_set_snd_timeout(prot->p_sfd, prot->p_settings.p_timeout, 0);

	if (prot->p_settings.p_type == PROTOCOL_TCP) { /* tcp */
		/* tcp set socket options */
		if (prot->p_settings.p_tcp_nodelay != 0) {
			(void)dserver_set_tcp_nodelay(prot->p_sfd);
		}
		if (prot->p_settings.p_tcp_keepalive != 0) {
			(void)dserver_set_tcp_keepalive(prot->p_sfd);
		}
		
		ret = dserver_tcp_conn_receive(prot);
		if (ret < 0) {
			DLOG_ERR(LOG_MAIN, "protocol[%s] running up error.", prot->p_name);
		}
	} else if (prot->p_settings.p_type == PROTOCOL_UDP) { /* udp */
		/* udp set socket options */
		ret = dserver_udp_data_receive(prot);
		if (ret < 0) {
			DLOG_ERR(LOG_MAIN, "protocol[%s] running up error.", prot->p_name);
		}
	} else { /* now not supported */
		DLOG_ERR(LOG_MAIN, "protocol type[%d] error.", prot->p_settings.p_type);
		ret = -E_PROT_UNSUPPORT;
	}

	return ret;
}

int reactor_init(dserver_t *dc)
{
	dserver_reactor_t *reactor;
	dserver_protocol_t *protocols;
	dserver_prot_node_t *tpos;
	list_node_t *pos;

	DLOG_DEBUG(LOG_MAIN, "dserver enters in reactor_init function.");

	if (dc == NULL || dc->settings == NULL || dc->protocols == NULL) {
		DLOG_ERR(LOG_MAIN, "reactor_init parameters error.");
		return -E_ARGS_ILLEGAL;
	}

	protocols = dc->protocols;
	reactor = dmalloc(sizeof(*reactor));
	dc->reactor = reactor;
	reactor->r_main_event = event_init();
	if (reactor->r_main_event == NULL) {
		DLOG_ERR(LOG_MAIN, "dserver main event create error in reactor.");
		return -E_EVENT;
	}

	/** create connection-pool, no matter whether has tcp-protocol-nodes */
	reactor->r_cpool = dserver_cpool_init(dc->settings->max_conns);
	if (reactor->r_cpool == NULL) {
		DLOG_ERR(LOG_MAIN, "dserver cpool create error in reactor.");
		return -E_NET_CONN;
	}

	/** link base struct to each protocol nodes */
	protocols->p_monitor->p_main_event = reactor->r_main_event;
	protocols->p_monitor->p_base.p_tcp_main_cpool = reactor->r_cpool;

	dlib_lock_lock(&protocols->p_lock);
	LIST_FOREACH_ENTRY(tpos, pos, &protocols->p_head, p_node) {
		tpos->p_prot.p_main_event = reactor->r_main_event;
		if (dserver_prot_type(&tpos->p_prot) == PROTOCOL_TCP) { /* tcp */
			tpos->p_prot.p_base.p_tcp_main_cpool = reactor->r_cpool;
		} else if (dserver_prot_type(&tpos->p_prot) == PROTOCOL_UDP) { /* udp */
			tpos->p_prot.p_base.p_udp_main_conn =
				dmalloc(sizeof(*tpos->p_prot.p_base.p_udp_main_conn));
			if (tpos->p_prot.p_base.p_udp_main_conn == NULL) {
				DLOG_ERR(LOG_MAIN, "dserver conn create error in reactor.");
				return -E_NET_CONN;
			} else {
				tpos->p_prot.p_base.p_udp_main_conn->c_prot_node = &tpos->p_prot;
			}
		}/* end prot_type */
	}
	dlib_lock_unlock(&protocols->p_lock);

	/** create notify_pipe_fd for pipeline finished reporting */
	if (pipe(reactor->r_done_pipe) < 0) {
		DLOG_ERR(LOG_MAIN, "pipe error in reactor: %s", strerror(errno));
		return -E_FILE_OPEN;
	}
	if (dserver_set_nonblock(reactor->r_done_pipe[0]) < 0) {
		DLOG_ERR(LOG_MAIN, "fd[%d] fcntl error in reactor: %s",
			 reactor->r_done_pipe[0], strerror(errno));
		return -E_FILE_FCNTL;
	}
	if (dserver_set_nonblock(reactor->r_done_pipe[1]) < 0) {
		DLOG_ERR(LOG_MAIN, "fd[%d] fcntl error in reactor: %s",
			 reactor->r_done_pipe[1], strerror(errno));
		return -E_FILE_FCNTL;
	}

	return RET_SUCCESS;
}

int reactor_run(dserver_t *dc)
{
	int ret = 0;
	dserver_reactor_t *reactor;
	dserver_protocol_t *protocols;
	dserver_prot_node_t *tpos;
	list_node_t *pos;
	
	if (dc == NULL || dc->reactor == NULL || dc->protocols == NULL) {
		DLOG_ERR(LOG_MAIN, "reactor_run parameters error.");
		return -E_ARGS_ILLEGAL;
	}

	reactor = dc->reactor;
	protocols = dc->protocols;

	/** add notify event callbacks */
	event_set(&reactor->r_pipe_event, reactor->r_done_pipe[0],
			EV_READ | EV_PERSIST, _reactor_pipe_handler, (void *)dc);
	(void)event_base_set(reactor->r_main_event, &reactor->r_pipe_event);
	ret = event_add(&reactor->r_pipe_event, 0);
	if (ret < 0) {
		DLOG_ERR(LOG_MAIN, "done_pipe event_add error.");
		return -E_EVENT;
	}
	
	/** running monitor protocol */
	if ((ret = _reactor_run_protocol(protocols->p_monitor)) < 0) {
		DLOG_ERR(LOG_MAIN, "reactor running monitor error.");
		return -E_UNEXPECTED;
	}
	
	/** run on all registered protocols */
	dlib_lock_lock(&protocols->p_lock);
	LIST_FOREACH_ENTRY(tpos, pos, &protocols->p_head, p_node) {
		if ((ret = _reactor_run_protocol(&tpos->p_prot)) < 0) {
			DLOG_ERR(LOG_MAIN, "running protocol[%s] error, abandon.",
					tpos->p_prot.p_name);
			continue;
		}
	}
	dlib_lock_unlock(&protocols->p_lock);

	/** dispatch, run into main loop */
	while (1) {
		ret = event_base_loop(reactor->r_main_event, 0);
		DLOG_ERR(LOG_MAIN, "event loop unexpected error[%d], loop again.", ret);
	}
}

void reactor_exit(dserver_t *dc)
{
	dserver_reactor_t *reactor;
	dserver_protocol_t *protocols;
	dserver_prot_node_t *tpos;
	list_node_t *pos;

	DLOG_DEBUG(LOG_MAIN, "dserver enters in reactor_exit function.");

	if (dc == NULL || dc->reactor == NULL || dc->protocols == NULL) {
		return;
	}
	reactor = dc->reactor;
	protocols = dc->protocols;
	
	dlib_lock_lock(&protocols->p_lock);
	LIST_FOREACH_ENTRY(tpos, pos, &protocols->p_head, p_node) {
		if (tpos->p_prot.p_sfd > 0 && tpos->p_prot.p_ops.p_server_exit != NULL) {
			(void)(*tpos->p_prot.p_ops.p_server_exit)(tpos->p_prot.p_sfd);
			tpos->p_prot.p_sfd = -1;
		}
		if (dserver_prot_type(&tpos->p_prot) == PROTOCOL_UDP && 
				tpos->p_prot.p_base.p_udp_main_conn != NULL) {
			dfree(tpos->p_prot.p_base.p_udp_main_conn);
			tpos->p_prot.p_base.p_udp_main_conn = NULL;
		}
	}
	dlib_lock_unlock(&protocols->p_lock);
	
	if (reactor->r_main_event != NULL) {
		event_base_free(reactor->r_main_event);
		reactor->r_main_event = NULL;
	}

	if (reactor->r_done_pipe[0] > 0) {
		close(reactor->r_done_pipe[0]);
		reactor->r_done_pipe[0] = -1;
	}
	if (reactor->r_done_pipe[1] > 0) {
		close(reactor->r_done_pipe[1]);
		reactor->r_done_pipe[1] = -1;
	}

	dserver_cpool_exit(reactor->r_cpool);
	dfree(reactor);

	return;
}
